package com.rui.study.micro.msgcenter.mq;

import com.rui.study.commons.model.BizException;
import com.rui.study.commons.util.JsonUtil;
import com.rui.study.micro.msgcenter.model.MailWrapper;
import com.rui.study.micro.msgcenter.service.mail.MailService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.context.config.annotation.RefreshScope;
import org.springframework.stereotype.Component;

import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @program: study
 * @description:
 * @author: Yaowr
 * @create: 2018-11-07 16:31
 **/
@Component
@RefreshScope
public class DefaultMessageListener {

    private static final Logger LOGGER = LoggerFactory.getLogger(DefaultMessageListener.class);

    private ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(4, new ThreadFactory() {
        private AtomicInteger num = new AtomicInteger();
        @Override
        public Thread newThread(Runnable r) {
            return new Thread(r, "Sending-" + num.getAndIncrement());
        }
    });

    @Autowired
    private MailService mailService;

    /**
     * ‘邮件通知’队列的监听器
     * @param message
     */
    @RabbitListener(
            bindings = @QueueBinding(
                    value = @Queue(
                            autoDelete = "true"
                    )
                    ,exchange = @Exchange(
                            value = "${msgcenter.mail.notification.exchange}",
                            type = ExchangeTypes.DIRECT,
                            autoDelete = "true"
                    )
                    ,key = "${msgcenter.mail.notification.routing_key_1}"
            )
            ,concurrency = "1")
    public void receive1(Message message) {
        MailWrapper mailWrapper;
        String body;
        try {
            body = new String(message.getBody(), "utf-8");
            LOGGER.info("Received a message from RabbitMq by receiver1: {}", body);
            mailWrapper = JsonUtil.toObject(body, MailWrapper.class);
            executor.execute(new DefaultMessageListener.SendingTask(mailWrapper));
            LOGGER.info("Executed the 'SendingTask'");
        } catch (Exception e) {
            LOGGER.error("系统异常", e);
        }
    }

    /**
     * 另一个‘邮件通知’队列的监听器
     * @param message
     */
    @RabbitListener(
            bindings = @QueueBinding(
                    value = @Queue(
                            autoDelete = "true"
                    )
                    ,exchange = @Exchange(
                            value = "${msgcenter.mail.notification.exchange}",
                            type = ExchangeTypes.DIRECT,
                            autoDelete = "true"
                    )
                    ,key = "${msgcenter.mail.notification.routing_key_2}"
            )
            ,concurrency = "1")
    public void receive2(Message message) {
        MailWrapper mailWrapper;
        String body;
        try {
            body = new String(message.getBody(), "utf-8");
            LOGGER.info("Received a message from RabbitMq by receiver2: {}", body);
            mailWrapper = JsonUtil.toObject(body, MailWrapper.class);
            executor.execute(new DefaultMessageListener.SendingTask(mailWrapper));
            LOGGER.info("Executed the 'SendingTask'");
        } catch (Exception e) {
            LOGGER.error("系统异常", e);
        }
    }


    private class SendingTask implements Runnable {

        private final Logger LOGGER = LoggerFactory.getLogger(DefaultMessageListener.SendingTask.class);

        private MailWrapper mailWrapper;

        private AtomicInteger count = new AtomicInteger();

        public SendingTask(MailWrapper mailWrapper) {
            this.mailWrapper = mailWrapper;
        }

        @Override
        public void run() {
            MailWrapper mailWrapper = this.mailWrapper;
            try {
                if (count.getAndIncrement() < 3) {
                    mailService.send(mailWrapper);
                } else {
                    LOGGER.warn("发送失败超过3次。。。: {}", JsonUtil.encode(mailWrapper));
                }
            } catch (BizException e) {
                String errMsg = e.getMessage();
                LOGGER.info("{}, 10秒后重发...\n{}", errMsg, mailWrapper);
                executor.schedule(this, 10, TimeUnit.SECONDS);
            }
        }
    }
}
